/*
 ***************************************************************************************
 * All rights Reserved, Designed By RTF
 * @Title:  RocketmqConsumerAdaptor.java   
 * @Package com.rtf.repair.mq.rocketmq.consumer
 * @Description: 消费者创建初始化  
 * @author: 司福林
 * @date:   2020-10-17 18:57:51   
 * @version V1.0 
 * @Copyright: 2020 RTF. All rights reserved.
 * 注意：本内容仅限于公司内部使用，禁止外泄以及用于其他的商业目
 *  ---------------------------------------------------------------------------------- 
 * 文件修改记录
 *     文件版本：         修改人：             修改原因：
 ***************************************************************************************
 */
package com.rtf.framework.mq.rocketmq.consumer;

import java.util.Map;
import java.util.Set;

import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.MessageListener;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.springframework.util.Assert;

import com.rtf.framework.common.util.UtilBeanFactory;
import com.rtf.framework.mq.common.BatchMessageConsumer;
import com.rtf.framework.mq.common.CommonConstants;
import com.rtf.framework.mq.common.ConsumerProperties;
import com.rtf.framework.mq.common.MQException;
import com.rtf.framework.mq.common.MessageConsumer;
import com.rtf.framework.mq.common.Service;
import com.rtf.framework.mq.rocketmq.consumer.listener.BatchMessageConsumerCurrentlyListener;
import com.rtf.framework.mq.rocketmq.consumer.listener.BatchMessageConsumerOrderListener;
import com.rtf.framework.mq.rocketmq.consumer.listener.MessageConsumerCurrentlyListener;
import com.rtf.framework.mq.rocketmq.consumer.listener.MessageConsumerOrderListener;
import com.google.common.collect.Maps;

/**
 *  
 * @Description: 消费者创建并初始化、启动开始消费
 * @ClassName:  RocketmqConsumerAdaptor
 * @author: 67077
 * @date:   2020-10-17 18:57:51
 * @since:  v1.0
 */
public class RocketmqConsumerAdaptor implements Service {

	/** The properties. */
	private RocketMqConsumerProperties properties;

	/** The consumer properties. */
	private ConsumerProperties consumerProperties;

	private DefaultMQPushConsumer consumer;

	/**
	 * Instantiates a new rocketmq consumer adaptor.
	 *
	 * @param properties the properties
	 * @param consumerProperties the consumer properties
	 */
	public RocketmqConsumerAdaptor(RocketMqConsumerProperties properties, ConsumerProperties consumerProperties) {
		Assert.notNull(properties, "RocketMqConsumerProperties is null");
		Assert.notNull(consumerProperties, "RocketMq cosunmer is null");

		this.properties = properties;
		this.consumerProperties = consumerProperties;
	}

	@Override
	public void init() {

		Assert.isTrue(StringUtils.isNotBlank(properties.getNamesrvAddr()), "Rocketmq consumer namesrvAddr is null");
		consumer = new DefaultMQPushConsumer(consumerProperties.getGroupName());
		consumer.setNamesrvAddr(properties.getNamesrvAddr());
		consumer.setInstanceName(consumerProperties.getInstanceName());
		consumer.setPollNameServerInterval(properties.getPollNameServerInterval());
		consumer.setHeartbeatBrokerInterval(properties.getHeartbeatBrokerInterval());
		consumer.setPersistConsumerOffsetInterval(properties.getPersistConsumerOffsetInterval());
		consumer.setConsumeThreadMin(properties.getConsumeThreadMin());
		consumer.setConsumeThreadMax(properties.getConsumeThreadMax());
		consumer.setConsumeMessageBatchMaxSize(properties.getConsumeMessageBatchMaxSize());
		consumer.setPullBatchSize(properties.getPullBatchSize());
		consumer.setMaxReconsumeTimes(properties.getMaxReconsumeTimes());
		consumer.setConsumeFromWhere(CommonConstants.CONSUMER_OFFSET_FIRST == consumerProperties.getConsumerWhere()
				? ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET
				: CommonConstants.CONSUMER_OFFSET_LAST == consumerProperties.getConsumerWhere()
						? ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET
						: ConsumeFromWhere.CONSUME_FROM_TIMESTAMP);

		if (CommonConstants.CONSUMER_OFFSET_TIMESTAMP == consumerProperties.getConsumerWhere()) {
			consumer.setConsumeTimestamp(UtilAll.timeMillisToHumanString3(consumerProperties.getConsumeTimestamp()));
		}
		MessageListener listener = null;
		Object handler = UtilBeanFactory.getApplicationContext().getBean(consumerProperties.getHandler());
		switch (consumerProperties.getListener()) {
			case RocketConsumerConstant.LISTENER_CUR:

				listener = new MessageConsumerCurrentlyListener((MessageConsumer) handler);
				break;

			case RocketConsumerConstant.LISTENER_CUR_BATCH:
				listener = new BatchMessageConsumerCurrentlyListener((BatchMessageConsumer) handler);
				break;

			case RocketConsumerConstant.LISTENER_ORDER:
				listener = new MessageConsumerOrderListener((MessageConsumer) handler);
				break;

			case RocketConsumerConstant.LISTENER_ORDER_BATCH:
				listener = new BatchMessageConsumerOrderListener((BatchMessageConsumer) handler);
				break;

			default:
				throw new MQException("创建RocketMq消费者监听器异常");

		}

		consumer.setMessageListener(listener);

		Set<String> topics = consumerProperties.getTopics();
		Assert.notEmpty(topics, "RocketMq consumer topic is empty");
		Map<String, String> topicMap = Maps.newHashMap();
		String[] str;
		for (String topic : topics) {
			str = topic.split("~");
			topicMap.put(str[0], str[1]);
		}
		consumer.setSubscription(topicMap);
	}

	@Override
	public void start() {
		try {
			consumer.start();
		} catch (MQClientException e) {
			throw new MQException("Rocketmq consumer starting failed", e);
		}
	}

	@Override
	public void destroy() {
		consumer.shutdown();
	}

}
